import pandas as pd
import numpy as np
from typing import List, Dict

class ContentRecommender:
    """
    基于内容的推荐器。
    根据物品（电影）的内容特征计算相似度来进行推荐。
    可以用于处理物品的冷启动问题或作为混合策略的一部分。
    """
    def __init__(self, movie_features_path: str, movies_metadata_path: str):
        """
        初始化内容推荐器。

        Args:
            movie_features_path (str): 包含电影数值化特征的pickle文件路径。
            movies_metadata_path (str): 包含电影元数据（如标题）的pickle文件路径。
        """
        # 加载特征与元数据
        movie_features = pd.read_pickle(movie_features_path)
        movies_meta = pd.read_pickle(movies_metadata_path)

        # 仅保留数值型特征列，并将其转换为 float32 以节省内存
        num_cols = [
            col for col in movie_features.columns
            if str(movie_features[col].dtype) in ['float32', 'float64', 'int32', 'int64', 'int', 'float']
        ]
        if 'movieId' not in num_cols:
            num_cols = ['movieId'] + num_cols

        feats_df = movie_features[num_cols].copy()
        feats_df = feats_df.set_index('movieId')
        # 将所有特征转换为 float32，缺失填 0
        # 优先使用 float32；如果内存仍然紧张，可考虑改为 float16（会有精度损失）
        try:
            feats_df = feats_df.astype('float32').fillna(0.0)
        except MemoryError:
            feats_df = feats_df.astype('float16').fillna(0.0)

        # 保存必要的结构，尽量使用 numpy 节省内存
        self.movie_ids = feats_df.index.to_numpy(dtype=np.int32)
        self.id_to_pos = {int(mid): i for i, mid in enumerate(self.movie_ids)}
        self.features = feats_df.to_numpy(dtype=np.float32, copy=False)
        # 预计算每行的 L2 范数，避免在每次查询时重复计算
        self.row_norms = np.linalg.norm(self.features, axis=1).astype(np.float32)
        self.row_norms[self.row_norms == 0] = 1.0  # 避免除零

        # 保存元数据（仅保留 title，避免不必要的内存占用）
        if 'movieId' in movies_meta.columns:
            movies_meta = movies_meta.set_index('movieId')
        self.movies_metadata = movies_meta

    def _compute_similarity_vector(self, movie_id: int, chunk_size: int = 8192) -> pd.Series | None:
        """
        计算指定电影与所有其他电影的余弦相似度（内存友好，分块计算）。
        使用自实现的余弦相似度，避免 sklearn 在内部将矩阵转换为 float64 并整体归一化带来的大内存占用。
        """
        pos = self.id_to_pos.get(int(movie_id))
        if pos is None:
            return None

        y = self.features[pos]
        y_norm = float(np.linalg.norm(y))
        if y_norm == 0:
            y_norm = 1.0

        n = self.features.shape[0]
        sims = np.empty((n,), dtype=np.float32)

        # 分块点积与归一化
        try:
            cur_chunk = int(chunk_size)
            while True:
                try:
                    for start in range(0, n, cur_chunk):
                        end = min(start + cur_chunk, n)
                        X = self.features[start:end]
                        # 点积
                        dots = X @ y  # (chunk,)
                        norms = self.row_norms[start:end]
                        denom = norms * y_norm
                        # 避免除零
                        denom = np.where(denom == 0, 1.0, denom)
                        sims[start:end] = (dots / denom).astype(np.float32)
                    break
                except MemoryError:
                    # 动态缩小分块，直到成功或最小为 512
                    if cur_chunk <= 512:
                        raise
                    cur_chunk = max(512, cur_chunk // 2)
        except MemoryError:
            # 在极端情况下返回空，交由上层兜底
            return pd.Series(dtype=np.float32)

        return pd.Series(sims, index=self.movie_ids)

    def recommend_similar_to(self, movie_id: int, top_k: int = 10) -> List[Dict]:
        """
        为给定电影返回最相似的 top_k 个电影（分块、低内存）。
        """
        sim_scores = self._compute_similarity_vector(movie_id)
        if sim_scores is None or sim_scores.empty:
            return []
        # 排除自身
        if movie_id in sim_scores.index:
            sim_scores.loc[movie_id] = -np.inf
        # 选取 top_k
        top = sim_scores.nlargest(top_k)
        recommendations: List[Dict] = []
        for m_id, score in top.items():
            title = ''
            try:
                if self.movies_metadata is not None and m_id in getattr(self.movies_metadata, 'index', []):
                    title = self.movies_metadata.loc[m_id, 'title'] if 'title' in self.movies_metadata.columns else ''
            except Exception:
                title = ''
            try:
                score_f = float(score)
            except Exception:
                score_f = 0.0
            recommendations.append({
                'movieId': int(m_id),
                'title': title,
                'similarity': score_f
            })
        return recommendations